package com.wl.mq;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Consumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.ShutdownSignalException;
import org.apache.commons.lang.SerializationUtils;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

/**
 * 读取队列的程序端，实现了Runnable接口
 * Created by zhen on 2016/10/11.
 */
public class QueueConsumer extends EndPoint implements Runnable, Consumer {

    public QueueConsumer(String endpointName) throws IOException {
        super(endpointName);
    }

    public void handleConsumeOk(String s) {}

    public void handleCancelOk(String s) {}

    public void handleCancel(String s) throws IOException {}

    public void handleShutdownSignal(String s, ShutdownSignalException e) {}

    public void handleRecoverOk(String consumerTag) {

        System.out.println("Consumer "+consumerTag +" registered");
    }

    public void handleDelivery(String s, Envelope envelope, AMQP.BasicProperties basicProperties, byte[] bytes) throws IOException {

        Map map = (HashMap) SerializationUtils.deserialize(bytes);
        System.out.println("Message Number "+ map.get("message number") + " received.");
    }

    public void run() {
        try {
            //start consuming messages. Auto acknowledge messages.
            channel.basicConsume(endPointName, true,this);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
